/**
 * Copyright (C), 2015-2018, XXX有限公司
 * FileName: myBolt
 * Author:   An-Il
 * Date:     2018/11/16 23:17
 * Description: 接受（处理）kafkaSpout发过来的数据
 * History:
 * <author>          <time>          <version>          <desc>
 * 作者姓名           修改时间           版本号              描述
 */
package com.blog.storm.example.integration.kafka.consumer;

import org.apache.storm.kafka.spout.DefaultRecordTranslator;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;

import java.util.Map;

/**
 * 〈一句话功能简述〉<br>
 * 〈接受（处理）kafkaSpout发过来的数据〉
 *
 * @author An-Il
 * @create 2018/11/16
 * @since 1.0.0
 */
public class myBolt extends BaseRichBolt {

    private OutputCollector collector;

    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
    }

    /**
     * new Fields("topic", "partition", "offset", "key", "value")
     *
     * @param input
     * @see DefaultRecordTranslator
     */
    @Override
    public void execute(Tuple input) {

        try {
            String topic = input.getStringByField("topic");
            Integer partition = input.getIntegerByField("partition");
            Long offset = input.getLongByField("offset");
            String key = input.getStringByField("key");
            String value = input.getStringByField("value");
            System.err.println("consumer message:" + value);

            this.collector.ack(input);
        } catch (Exception e) {
            this.collector.fail(input);
            e.printStackTrace();
        }

    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {

    }
}